package com.itdl.mq.config;

import com.itdl.cache.config.AbstractRedisConfig;
import com.itdl.cache.config.CustomRedisConfig;
import com.itdl.cache.manager.CacheNullValuesHandle;
import com.itdl.common.base.properties.CustomStreamMqProperties;
import com.itdl.mq.base.AbstractStreamConsumerListener;
import com.itdl.mq.base.BaseFanoutMessageListener;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.AutowireCapableBeanFactory;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.boot.autoconfigure.AutoConfigureBefore;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.util.CollectionUtils;

import java.time.Duration;
import java.util.Collections;
import java.util.List;

/**
 * 自定义Redis MQ配置文件
 */
@Configuration  // 标识为一个配置项，注入Spring容器
@AutoConfigureBefore({CustomRedisConfig.class, CacheNullValuesHandle.class, AbstractStreamConsumerListener.class, AutowireCapableBeanFactory.class})
@EnableConfigurationProperties({CustomStreamMqProperties.class}) // 启动Redis配置文件
@ConditionalOnProperty(value = "redis.enable", havingValue = "true")
@Slf4j
@SuppressWarnings("all")
public class RedisMqConfig extends AbstractRedisConfig implements ApplicationContextAware {

    // 消息队列和监听器绑定的配置文件
    @Autowired
    private CustomStreamMqProperties streamMqProperties;

    // 发布订阅模式消息监听器列表（继承BaseFanoutMessageListener注入spring容器，这里就能获取到）
    @Autowired(required = false)
    private List<BaseFanoutMessageListener> messageListeners;

    // 点对点模式消息监听器列表（继承StreamListener注入spring容器，这里就能获取到）
    @Autowired(required = false)
    private List<StreamListener> streamListeners;

    // spring容器上下文，用于获取Bean
    private ApplicationContext applicationContext;

    /**
     * 配置Stream消息监听容器Options
     */
    @Bean
    public StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String,String,String>> streamMessageListenerContainerOptions(){
        StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();

        return StreamMessageListenerContainer.StreamMessageListenerContainerOptions
                .builder()
                // block读取超时时间
                .pollTimeout(Duration.ofSeconds(30))
                // count 数量（一次只获取一条消息）
                .batchSize(1)
                // 序列化规则
                .serializer(stringRedisSerializer)
                .build();
    }

    /**
     * 配置stream监听器接收消息
     */
    @Bean
    public StreamMessageListenerContainer<String,MapRecord<String,String,String>> streamMessageListenerContainer(RedisConnectionFactory factory,
                                                                                                                 StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String,String,String>> streamMessageListenerContainerOptions,
                                                                                                                 RedisTemplate<String, Object> redisTemplate){

        StreamMessageListenerContainer<String,MapRecord<String,String,String>> listenerContainer = StreamMessageListenerContainer.create(factory,
                streamMessageListenerContainerOptions);


        final List<CustomStreamMqProperties.StreamConfig> configs = streamMqProperties.getConfigs();

        // 如果配置为空或者容器中没有监听器注入，则不启动
        if (CollectionUtils.isEmpty(configs) || CollectionUtils.isEmpty(streamListeners)){
            return listenerContainer;
        }

        // 遍历队列和监听器配置列表
        for (CustomStreamMqProperties.StreamConfig config : configs) {
            //如果 流不存在 创建 stream 流
            String streamName = config.getQueueName() + "Stream";
            // 组名
            String groupName = config.getQueueName() + "Group";

            // 如果stream不存在，则创建
            if(!redisTemplate.hasKey(streamName)){
                redisTemplate.opsForStream().add(streamName, Collections.singletonMap("", ""));
                log.info("======>>>初始化stream {} success", streamName);
            }

            //创建消费者组名不存在，则创建
            try {
                redisTemplate.opsForStream().createGroup(streamName, groupName);
                log.info("======>>>初始化group {} success", groupName);
            } catch (Exception e) {
                log.warn("======>>>消费者组 {} 已存在", groupName);
            }

            // 获取消费者监听器的全限定类名
            String messageListenerFullName = config.getMessageListenerFullName();

            // 使用自定义的消息监听器和stream进行绑定。
            // 遍历所有的注入容器中的stream监听器
            for (StreamListener messageListener : streamListeners) {
                // 如果监听器名称和配置全限定名称一样
                if (messageListener.getClass().getName().equalsIgnoreCase(messageListenerFullName.trim())){
                    // 则将监听器和stream和消费者组进行绑定，从而接收消息
                    listenerContainer.receive(
                            Consumer.from(groupName, config.getQueueName() + "Consumer"),
                            StreamOffset.create(streamName, ReadOffset.lastConsumed()),
                            messageListener
                    );
                    log.info("add stream listener {} success", messageListener.getClass().getName());
                }
            }
        }

        // 所以的stream和监听器绑定之后，启动strean监听器
        listenerContainer.start();
        return listenerContainer;
    }



    /**
     * redis监听器配置（发布订阅模式）
     */
    @Bean
    public RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory) {
        // 创建一个Redis消息监听容器
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        // 将连接工厂设置到容器中
        container.setConnectionFactory(connectionFactory);
        // 配置广播/发布订阅模式消息监听器
        List<CustomStreamMqProperties.TopicConfig> topics = streamMqProperties.getTopics();

        // not start
        if (CollectionUtils.isEmpty(topics) || CollectionUtils.isEmpty(messageListeners)){
            return container;
        }

        // 果配置为空或者容器中没有监听器注入，则不启动
        for (CustomStreamMqProperties.TopicConfig topic : topics) {
            // 获取topic名称
            String topicName = topic.getTopicName().trim();
            // 获取消息监听器的全限定类名
            String messageListenerFullName = topic.getMessageListenerFullName();
            // 使用自定义的消息监听器
            for (BaseFanoutMessageListener messageListener : messageListeners) {
                // 如果当前消息监听器名称就是配置的名称
                if (messageListener.getClass().getName().equalsIgnoreCase(messageListenerFullName.trim())){
                    // 手动注册一个消息监听适配器。名称以topic + "ListenerAdapter"为适配器名称
                    MessageListenerAdapter adapter = (MessageListenerAdapter)registerBean(topic + "ListenerAdapter", listenerAdapter(messageListener));
                    log.info("register {} mapping MessageListenerAdapter", messageListener.getClass().getName());
                    container.addMessageListener(adapter, new PatternTopic(topicName));
                }
            }
        }

        return container;
    }


    // 根据见日你刚起创建一个监听器适配器
    public MessageListenerAdapter listenerAdapter(MessageListener listener) {
        MessageListenerAdapter listenerAdapter = new MessageListenerAdapter(listener, "onMessage");
        // 配置序列化器
        listenerAdapter.setSerializer(buildRedisJackson());
        return listenerAdapter;
    }


    // 手动注册Bean，因为需要AutowireCapableBeanFactory进行注册bean, 所以需要使用@AutoConfigureBefore({AutowireCapableBeanFactory.class}) 提前加载该bean
    public Object registerBean(String beanName, Object singletonObject){
        DefaultListableBeanFactory defaultListableBeanFactory = (DefaultListableBeanFactory) applicationContext.getAutowireCapableBeanFactory();

        //动态注册bean.
        defaultListableBeanFactory.registerSingleton(beanName, singletonObject);

        //获取动态注册的bean.
        return applicationContext.getBean(beanName);
    }

    // 设置上下文，因为DefaultListableBeanFactory是从上下文中获取的
    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }
}
